package com.mgq.java8.future;

import java.util.Arrays;
import java.util.List;
import java.util.Random;
import java.util.concurrent.*;
import java.util.stream.Collectors;
import java.util.stream.Stream;

public class Shop {

    private String name;
    private static Random random = new Random();
    private static List<Shop> shops = Arrays.asList(new Shop("taoBao"),
            new Shop("jingDong"),
            new Shop("pinDuoDuo"),
            new Shop("Amazon"),
            new Shop("tianMao"),
            new Shop("DangDang"),
            new Shop("AppleStore"),
            new Shop("WeiPinHui"));

    public Shop(String name) {
        this.name = name;
    }

    public String getName() {
        return name;
    }

    public void setName(String name) {
        this.name = name;
    }

    private static final ExecutorService executorService = Executors.newFixedThreadPool(Math.min(shops.size(), 100), new ThreadFactory() {
        int count = 0;

        @Override
        public Thread newThread(Runnable r) {
            Thread t = new Thread(r, "获取价格线程:" + count);
            count++;
            //设置为守护线程
            t.setDaemon(true);
            return t;
        }
    });

    /**
     * 同步方法
     */
    public double getPrice(String product) {

        // 该方法的内部实现会查询商店的数据库，但也有可能执行一些其他耗时的任务，比如联系其
        //他外部服务（比如，商店的供应商，或者跟制造商相关的推广折扣）。
        //使用sleep来模拟延迟
        return calculatePrice(product);
    }


    /**
     * 使用CompleteFuture来实现异步(jdk1.8)
     * 如何正确地管理异步任务执行过程中可能出现的错误。使用completeExceptionally(e)方法
     * 使用CompletableFuture提供的supplyAsync()方法也可以实现错误.当发生异常时,会将异常抛出.
     * supplyAsync()这个方法,相当于getPriceAsync()方法
     */
    public Future<Double> getPriceAsync(String product) {
        //创建CompletableFuture对象,它会包含计算结果
        CompletableFuture<Double> futurePrice = new CompletableFuture<>();
        //在另外一个线程计算结果
        new Thread(() -> {
            try {
                //如果价格计算正常结束，完成Future操作并设置商品价格
                double price = calculatePrice(product);
                //需长时间计算的任务结束并得出结果时，设置Future的返回值
                futurePrice.complete(price);
                //任务完成时,发送通知
                //notify()
            } catch (Exception e) {
                //使用completeExceptionally()方法,将异常抛出.这样主线程就知道是什么异常引发的
                //抛出CompletableFuture内的异常
                futurePrice.completeExceptionally(e);
            }
        }).start();
        return futurePrice;
    }

    /**
     * 使用工厂方法创建CompletableFuture(jdk1.8) 这个方法完全等价于@{@link Shop#getPriceAsync}
     * supplyAsync方法接受一个生产者（Supplier）作为参数，返回一个CompletableFuture
     * 对象，该对象完成异步执行后会读取调用生产者方法的返回值。生产者方法会交由ForkJoinPool
     * 池中的某个执行线程（Executor）运行，但是你也可以使用supplyAsync方法的重载版本，传
     * 递第二个参数指定不同的执行线程执行生产者方法。
     */
    public Future<Double> getPriceUseFactoryAsync(String product) {
        return CompletableFuture.supplyAsync(() -> calculatePrice(product));

    }

    //改造getPrice()返回 name:price:code 如百度:163.61:SILVER
    public String getPriceWithCode(String product) {
        double price = calculatePrice(product);
        Discount.Code code = Discount.Code.values()[random.nextInt((Discount.Code.values().length))];
        return String.format("%s:%.2f:%s", name, price, code);
    }

    public static void main(String[] args) {
        testCompletableFuture();
        System.out.println(Runtime.getRuntime().availableProcessors());
        long start = System.nanoTime();
        //System.out.println(findPrices("iphone13"));
        //使用并行流的效果
        //List<String> iphone13 = findPricesParallel("iphone13");
        //System.out.println();
        //使用completableFuture提供是supplyAsync()方法来并行执行
//        List<CompletableFuture<String>> iphone13 = findPricesUserCompletableFuture("iphone13");
       // List<String> iphone13 = findPricesUserCompletableFutureJoin("iphone13");
        //使用自定义的线程池
        List<String> iphone13 = findPriceUseCompletableFutureThreadPool("iphone13");
        System.out.println(iphone13);
        long duration = (System.nanoTime() - start) / 1_000_000;
        System.out.println("最终用时: " + duration + " msecs");
    }

    //让代码免受阻塞之苦
    //同步调用, 最终用时: 4040 msecs
    public static List<String> findPrices(String product) {
        return shops.stream()
                .map(shop -> String.format("%s 价格是 %.2f", shop.getName(), shop.getPrice(product)))
                .collect(Collectors.toList());
    }

    //尝试使用并行流
    //最终用时: 1031 msecs.
    public static List<String> findPricesParallel(String product) {
        return shops.stream()
                .parallel()
                .map(shop -> String.format("%s 价格是 %.2f", shop.getName(), shop.getPrice(product)))
                .collect(Collectors.toList());
    }

    //使用completableFuture提供的supplyAsync()方法来发起异步请求
    public static List<CompletableFuture<String>> findPricesUserCompletableFuture(String product) {
        List<Shop> shops = Arrays.asList(new Shop("taoBao"),
                new Shop("jingDong"),
                new Shop("pinDuoDuo"),
                new Shop("Amazon"));
        return shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(() -> String.format("%s 价格是%.2f", shop.getName(), shop.getPrice(product))))
                .collect(Collectors.toList());
    }

    //由于findPricesUserCompletableFuture()方法返回的List<CompletableFuture<String>>
    //我们最终需要的是List<String>的结果,所以使用CompletableFuture提供的join()方法将最终得到的结果连接起来
    //join()方法不会抛出出任何检测到的异常,因此不需要try-catch,而get()方法会抛出异常,需要catch
    //join()方法是为了更好的结合函数式调用.
    //此处使用了2个流的原因是为了避免 由于流的延迟特性会引起顺序执行
    //使用2个的原因,其实是supplyAsync()方法会立即返回. 而如果用一个流的2个map,会导致阻塞在join()方法.导致流最终串行化
    public static List<String> findPricesUserCompletableFutureJoin(String product) {
        List<Shop> shops = Arrays.asList(new Shop("taoBao"),
                new Shop("jingDong"),
                new Shop("pinDuoDuo"),
                new Shop("Amazon"));

        //使用一个流,无法并行化
        /*return  shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(() -> String.format("%s 价格是%.2f", shop.getName(), shop.getPrice(product))))
                .map(CompletableFuture::join)
                .collect(Collectors.toList());*/

        List<CompletableFuture<String>> priceFutures = shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(() -> String.format("%s 价格是%.2f", shop.getName(), shop.getPrice(product))))
                .collect(Collectors.toList());

        //使用join()方法将返回结果组装起来
        return priceFutures.stream().map(CompletableFuture::join).collect(Collectors.toList());
    }


    //调整线程池的大小
    //《Java并发编程实战》（http://mng.bz/979c）一书中，Brian Goetz和合著者们为线程池大小
    //的优化提供了不少中肯的建议。这非常重要，如果线程池中线程的数量过多，最终它们会竞争
    //稀缺的处理器和内存资源，浪费大量的时间在上下文切换上。反之，如果线程的数目过少，正
    //如你的应用所面临的情况，处理器的一些核可能就无法充分利用。Brian Goetz建议，线程池大
    //小与处理器的利用率之比可以使用下面的公式进行估算：
    //Nthreads = NCPU * UCPU * (1 + W/C)
    //其中：
    //❑NCPU是处理器的核的数目，可以通过Runtime.getRuntime().availableProcessors()得到  16
    //❑UCPU是期望的CPU利用率（该值应该介于0和1之间）                                    100
    //❑W/C是等待时间与计算时间的比率                                                  99

    //比如: 我想让CPU的利用率达到100%, 99%的时间都在等待,计算花费了1%, 那么需要创建16 * 100 =1600个线程,才能达到100%的CPU利用率

    //使用自定义的线程池来处理任务,由于并行流内部的线程数个数是Runtime.getRuntime().availableProcessors().当我们需要自定义线程个数的时候,
    //并行流无法做到,而CompletableFuture可以使用自定义的线程池. 这样可以设置自己的线程池大小,比如我shop的size大于16时,使用并行流会达到2秒
    //而使用CompletableFuture.supplyAsync()时,结果仍然是1秒.(线程数大于16)
    public static List<String> findPriceUseCompletableFutureThreadPool(String product) {
        List<CompletableFuture<String>> collect = shops.stream()
                .map(shop -> CompletableFuture
                        //executorService使用指定的线程池
                        .supplyAsync(() -> String.format("%s 价格是%.2f", shop.getName(), shop.getPrice(product)), executorService))
                .collect(Collectors.toList());
        return collect.stream().map(CompletableFuture::join).collect(Collectors.toList());
    }

    //1.应用现在能从不同的商店取得商品价格.
    //2.解析结果字符串，针对每个字符串，查询折扣服务取的折扣代码

    // 第一个操作将每个shop对象转换成了一个字符串，该字符串包含了该 shop中指定商品的价格和折扣代码。
    // 第二个操作对这些字符串进行了解析，在Quote对象中对它们进行转换。
    // 最终，第三个map会操作联系远程的Discount服务，计算出最终的折扣价格，并返回该价格及提供该价格商品的shop。
    public static List<String> findPricesDiscount(String product) {
        //共有8家,最终返回的结果是16s
        return shops.stream()
                .map(shop -> shop.getPriceWithCode(product))
                .map(Quote::parse)
                .map(Discount::applyDiscount)
                .collect(Collectors.toList());
    }

    //thenApply()   转换的是泛型中的类型，相当于将CompletableFuture<T> 转换生成新的CompletableFuture<U>
    //thenCompose() 方法允许你对两个异步操作进行流水线，第一个操作完成时，将其结果作为参数传递给第二个操作。换句话说，你可以创建两个CompletableFutures对象，对
    //第一个CompletableFuture对象调用 thenCompose，并向其传递一个函数。当第一个CompletableFuture执行完毕后，它的结果将作为该函数的参数，这个函数的返回值是以第一
    //个CompletableFuture的返回做输入计算出的第二个CompletableFuture对象。
    //如果你需要将两个完全不相干的CompletableFuture对象的结果整合起来，而且你也不希望等到第一个任务完全结束才开始第二项任务。这种情况，你应该使用thenCombine方法

    //thenCombine() 会在两个任务都执行完成后，把两个任务的结果合并。
    //注意:
    //两个任务中只要有一个执行异常，则将该异常信息作为指定任务的执行结果。
    //两个任务是并行执行的，它们之间并没有先后依赖顺序。
    public static List<String> findPriceUseCompletableFuture(String product) {
        List<CompletableFuture<String>> collect = shops.stream()
                //以异步的方式去得到每个商店的产品的价格
                .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPriceWithCode(product), executorService))
                //Quote对象存在时，对其返回的值进行转换
                .map(future -> future.thenApply(Quote::parse))
                //使用另一个任务构造期望的future.申请折扣 (相当于再次使用异步)
                .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(
                        () -> Discount.applyDiscount(quote), executorService))
                )
                .collect(Collectors.toList());

        //等待流中的所有Future执行完毕，并提取各自的返回值
        return collect.stream()
                .map(CompletableFuture::join)
                .collect(Collectors.toList());
    }

    //使用thenCombine: 查找汇率和获取价格是2个不相关的事情,没有先后顺序,最终只需要计算最终价格就行
    public static List<String> findPricesCalculateRate(String product) {
        List<CompletableFuture<Double>> collect = shops.stream()
                .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPrice(product))
                        .thenCombine(CompletableFuture.supplyAsync(Shop::getRate), (price, rate) -> price * rate))
                .collect(Collectors.toList());
        return collect.stream().map(CompletableFuture::join).map(price->String.format("%.2f",price)).collect(Collectors.toList());
    }

    //使用jdk7实现thenCombine()功能
    public static String findPriceUserJdk7(String product) throws ExecutionException, InterruptedException {
        ExecutorService executorService = Executors.newCachedThreadPool();
        Future<Double> rateFuture = executorService.submit(new Callable<Double>() {
            @Override
            public Double call() throws Exception {
                return Shop.getRate();
            }
        });
        Future<String> finalPrice = executorService.submit(new Callable<String>() {
            @Override
            public String call() throws Exception {
                Double rate = rateFuture.get();
                double price = new Shop("taobao").getPrice(product);
                return String.format("%.2f", rate * price);
            }
        });
        return finalPrice.get();
    }

    //查询汇率
    public static double getRate() {
        delay();
        return 6.67d;
    }

    //响应 CompletableFuture 的 completion 事件

    //实时的获取价格,不再一直等待所有的完成
    //Java 8的CompletableFuture通过thenAccept方法提供了这一功能，它接收CompletableFuture执行完毕后的返回值做参数。
    //findPriceRealTime("iphone13").forEach(f -> f.thenAccept(System.out::println));
    public static Stream<CompletableFuture<String>> findPriceRealTime(String product) {
        return shops.stream()
                //以异步的方式去得每个商店的产品的价格
                .map(shop -> CompletableFuture.supplyAsync(() -> shop.getPriceWithCode(product), executorService))
                //Quote对象存在时，对其返回的值进行转换
                .map(future -> future.thenApply(Quote::parse))
                //使用另一个任务构造期望的future.申请折扣
                .map(future -> future.thenCompose(quote -> CompletableFuture.supplyAsync(
                        () -> Discount.applyDiscount(quote), executorService))
                );
    }





    /*public static void main(String[] args) {
        String result = CompletableFuture.supplyAsync(() -> {
            System.out.println("开始异步执行,把异步结果hello返回");
            return "hello";
        }).thenApply((s) -> {
            System.out.println("接收到的字符串是:" + s);
            return "world";
        }).thenCompose(
                s -> {
                    System.out.println("第二次接收到的字符串是:" + s);
                    return CompletableFuture.supplyAsync(() -> s + "连接完成");
                }).join();
        System.out.println(result);
    }*/

  /*  public static void main(String[] args) throws InterruptedException {
        long start = System.nanoTime();
//        List<String> iphone13 = findPricesDiscount("iphone13");
       //List<String> iphone13 = findPriceUseCompletableFuture("iphone13");
        //System.out.println(iphone13);
        //List<String> iphone13 = findPricesCalculateRate("iphone13");
        //findPriceRealTime("iphone13").forEach(f -> f.thenAccept(System.out::println));

        CompletableFuture[] futures = findPriceRealTime("iphone13")
                .map(f -> f.thenAccept(s -> System.out.println(s + " (done in " +
                        ((System.nanoTime() - start) / 1_000_000) + " msecs)")))
                .toArray(CompletableFuture[]::new);
        CompletableFuture.allOf(futures).join();

        System.out.println("All shops have now responded in "
                + ((System.nanoTime() - start) / 1_000_000) + " msecs");
        *//*System.out.println(iphone13);
        long duration = (System.nanoTime() - start) / 1_000_000;
        System.out.println("最终用时: " + duration + " msecs");*//*
    }*/

    //并行流和CompletableFuture该怎么选择?
    //1.如果任务是计算密集型的,使用Stream流
    //2.如果任务由IO,并且IO时间比较长,推荐使用CompletableFuture.因为可以调整线程数

    /**
     * <b>功能描述：</b>使用CompletableFuture<br>
     * <b>修订记录：</b><br>
     * <li>20220609&nbsp;&nbsp;|&nbsp;&nbsp;马广奇&nbsp;&nbsp;|&nbsp;&nbsp;创建方法</li><br><br>
     *
     * @param
     * @return void
     */
    private static void testCompletableFuture() {
        Shop shop = new Shop("best shop");
        long start = System.nanoTime();
        //Future<Double> futurePrice = shop.getPriceAsync("my favorite product");
         Future<Double> futurePrice = shop.getPriceUseFactoryAsync("my favorite product");
        long invocationTime = ((System.nanoTime() - start) / 1_000_000);
        System.out.println("调用返回后用时 " + invocationTime
                + " msecs");

        // 执行更多任务，比如查询其他商店
        doSomethingElse();
        try {
            //从future中获取价格,如果价格未知,则等待
            //正常应该使用get(timeout,unit)方法来避免无限制等待.
            //但是无法获取到具体的异常.
            Double price = futurePrice.get(2, TimeUnit.SECONDS);
            System.out.printf("价格是 %.2f%n", price);
        } catch (InterruptedException | ExecutionException | TimeoutException e) {
            throw new RuntimeException(e);
        }
        //恢复时间
        long retrievalTime = ((System.nanoTime() - start) / 1_000_000);
        System.out.println("价格返回后用时 " + retrievalTime + " msecs");
    }

    private static void doSomethingElse() {
        System.out.println("执行其他事情");

    }

    /**
     * 计算商品价格
     */
    private double calculatePrice(String product) {

        //delay();
        randomDelay();
         int i = 2 / 0;
        return random.nextDouble() * product.charAt(0) + product.charAt(1);
    }

    /**
     * 模拟延迟操作
     */
    public static void delay() {
        try {
            Thread.sleep(1000L);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }

    //生成一个随机的延迟 0.5-2.5秒之间
    public static void randomDelay() {
        int delay = 500 + random.nextInt(2000);
        try {
            Thread.sleep(delay);
        } catch (InterruptedException e) {
            throw new RuntimeException(e);
        }
    }
}
